package com.alibaba.datax.plugin.writer.dmwriter;

import com.alibaba.datax.common.util.Configuration;
import org.apache.commons.lang3.StringUtils;

import java.util.List;
import java.util.Map;

/**
 * DataX Job 配置动态生成器
 */
public class DmJobConfigGenerator {

    /**
     * 生成包含主键信息的 DataX Job 配置
     */
    public static Configuration generateJobWithPrimaryKey(
            String readerUrl, String readerUsername, String readerPassword,
            String writerUrl, String writerUsername, String writerPassword,
            List<String> tables) {

        DmMetadataUtil metadataUtil = null;
        try {
            // 创建配置访问达梦数据库
            Configuration dmConfig = Configuration.newDefault();
            dmConfig.set("jdbcUrl", readerUrl);
            dmConfig.set("username", readerUsername);
            dmConfig.set("password", readerPassword);

            metadataUtil = new DmMetadataUtil(dmConfig);

            // 获取所有表的主键信息
            Map<String, List<String>> allPrimaryKeys = metadataUtil.getBatchPrimaryKeys(tables);

            // 生成 DataX Job 配置
            return buildDataXJobConfig(readerUrl, readerUsername, readerPassword,
                    writerUrl, writerUsername, writerPassword, tables, allPrimaryKeys);

        } finally {
            if (metadataUtil != null) {
                metadataUtil.close();
            }
        }
    }

    private static Configuration buildDataXJobConfig(
            String readerUrl, String readerUsername, String readerPassword,
            String writerUrl, String writerUsername, String writerPassword,
            List<String> tables, Map<String, List<String>> primaryKeys) {

        Configuration jobConfig = Configuration.newDefault();

        // 构建多个 content（每个表一个）
        for (String table : tables) {
            Configuration content = buildTableContent(
                    table, primaryKeys.get(table),
                    readerUrl, readerUsername, readerPassword,
                    writerUrl, writerUsername, writerPassword
            );

            jobConfig.getList("job.content", Configuration.class).add(content);
        }

        // 设置公共配置
        jobConfig.set("job.setting.speed.channel", tables.size());

        return jobConfig;
    }

    private static Configuration buildTableContent(
            String table, List<String> primaryKeys,
            String readerUrl, String readerUsername, String readerPassword,
            String writerUrl, String writerUsername, String writerPassword) {

        Configuration content = Configuration.newDefault();

        // Reader 配置
        Configuration reader = Configuration.newDefault();
        reader.set("name", "dmreader");
        reader.set("parameter.username", readerUsername);
        reader.set("parameter.password", readerPassword);
        reader.set("parameter.connection[0].jdbcUrl[0]", readerUrl);
        reader.set("parameter.connection[0].table[0]", table);

        // 如果有主键，设置分片键
        if (primaryKeys != null && !primaryKeys.isEmpty()) {
            reader.set("parameter.splitPk", primaryKeys.get(0));
        }

        // Writer 配置
        Configuration writer = Configuration.newDefault();
        writer.set("name", "dmwriter");
        writer.set("parameter.username", writerUsername);
        writer.set("parameter.password", writerPassword);
        writer.set("parameter.connection[0].jdbcUrl", writerUrl);
        writer.set("parameter.connection[0].table[0]", table);

        // 如果有主键，设置 MERGE INTO 语句
        if (primaryKeys != null && !primaryKeys.isEmpty()) {
            String mergeSql = generateMergeSql(table, primaryKeys);
            writer.set("parameter.preSql[0]", mergeSql);
        }

        content.set("reader", reader);
        content.set("writer", writer);

        return content;
    }

    private static String generateMergeSql(String table, List<String> primaryKeys) {
        StringBuilder sql = new StringBuilder();
        sql.append("MERGE INTO ").append(table).append(" target USING (SELECT ");

        // 假设列名已知，这里需要根据实际情况调整
        // 这里只是示例，实际使用时需要获取所有列名
        sql.append("? as id, ? as name, ? as age") // 示例列
                .append(" FROM dual) source ON (");

        // 构建 ON 条件（主键条件）
        for (int i = 0; i < primaryKeys.size(); i++) {
            if (i > 0) sql.append(" AND ");
            sql.append("target.").append(primaryKeys.get(i))
                    .append(" = source.").append(primaryKeys.get(i));
        }

        sql.append(") WHEN MATCHED THEN UPDATE SET target.name=source.name, target.age=source.age ")
                .append("WHEN NOT MATCHED THEN INSERT (id, name, age) VALUES (source.id, source.name, source.age)");

        return sql.toString();
    }
}
